#include "mixp_threadpool.h"
#include "core/log/mixp_log.h"

typedef enum
{
    IMMEDIATE = 1,
    GRACEFUL = 2
} mixp_threadpool_shutdown_t;

typedef struct mixp_thread_task
{
    void (*func)(void *);
    void *args;
} mixp_thread_task_t;

static void *thrd_task_executor(void *);
int mixp_threadpool_free(mixp_thread_pool_t *);

mixp_thread_pool_t *mixp_threadpool_new(int n_threads, int queue_size)
{
    if (n_threads <= 0 || n_threads > MAX_THREADS)
    {
        loge("n_threads must be in %d - %d", 0, MAX_THREADS);
        return NULL;
    }

    mixp_thread_pool_t *pool = MALLOC(mixp_thread_pool_t);
    if (pool == NULL)
    {
        loge("malloc thread pool failed.");
        goto err;
    }

    pool->n_threads = 0;
    pool->size = (queue_size > 0 && queue_size < MAX_THREAD_TASK_QUEUE_SIZE) ? queue_size : MAX_THREAD_TASK_QUEUE_SIZE;
    pool->head = pool->tail = pool->count = 0;
    pool->shutdown = pool->started = 0;

    pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * n_threads);
    pool->queue = (mixp_thread_task_t *)malloc(sizeof(mixp_thread_task_t) * pool->size);
    if (pool->threads == NULL || pool->queue == NULL)
    {
        loge("init threads or queue failed. threads: %p, queue: %p", pool->threads, pool->queue);
        goto err;
    }

    if (pthread_mutex_init(&(pool->lock), NULL) != 0 || pthread_cond_init(&(pool->notify), NULL) != 0)
    {
        loge("init lock failed.");
        goto err;
    }

    for (int i = 0; i < n_threads; i++)
    {
        if (pthread_create(&(pool->threads[i]), NULL, thrd_task_executor, (void *)pool) != 0)
        {
            mixp_threadpool_delete(pool, 0);
            return NULL;
        }

        pool->n_threads++;
        pool->started++;
    }

    return pool;

err:
    if (pool)
    {
        FREE(pool->threads);
        FREE(pool->queue);
        FREE(pool);
    }
    return NULL;
}

int mixp_threadpool_add_task(mixp_thread_pool_t *pool, void (*routine)(void *), void *args, int flags)
{
    int err = 0;
    int next;

    if (pool == NULL || routine == NULL)
    {
        return INVALID;
    }

    if (pthread_mutex_lock(&(pool->lock)) != 0)
    {
        return LOCK_FAILED;
    }

    next = pool->tail + 1;
    next = (next == pool->size) ? 0 : next;

    do
    {
        if (pool->count == pool->size)
        {
            err = QUEUE_IS_FULL;
            break;
        }

        if (pool->shutdown)
        {
            err = SHUTDOWN;
            break;
        }

        pool->queue[pool->tail].func = routine;
        pool->queue[pool->tail].args = args;

        pool->tail = next;
        pool->count += 1;

        if (pthread_cond_signal(&(pool->notify)) != 0)
        {
            err = LOCK_FAILED;
            break;
        }

    } while (0);

    if (pthread_mutex_unlock(&(pool->lock)) != 0)
    {
        err = LOCK_FAILED;
    }

    return err;
}

int mixp_threadpool_delete(mixp_thread_pool_t *pool, int flags)
{
    int err = 0;

    if (pool == NULL)
    {
        return INVALID;
    }

    if (pthread_mutex_lock(&(pool->lock)) != 0)
    {
        return LOCK_FAILED;
    }

    do
    {
        if (pool->shutdown)
        {
            err = SHUTDOWN;
            break;
        }

        pool->shutdown = (flags & GRACEFULLY) ? GRACEFUL : SHUTDOWN;

        if ((pthread_cond_broadcast(&(pool->notify)) != 0) || (pthread_mutex_unlock(&(pool->lock)) != 0))
        {
            err = LOCK_FAILED;
            break;
        }

        for (int i = 0; i < pool->n_threads; i++)
        {
            if (pthread_join(pool->threads[i], NULL) != 0)
            {
                err = THREAD_FAILTURE;
                logw("err: %d", err);
            }
        }
    } while (0);

    if (!err)
    {
        mixp_threadpool_free(pool);
    }

    return err;
}

int mixp_threadpool_free(mixp_thread_pool_t *pool)
{
    if (pool == NULL || pool->started > 0)
    {
        return -1;
    }

    FREE(pool->threads);
    FREE(pool->queue);

    pthread_mutex_lock(&(pool->lock));
    pthread_mutex_destroy(&(pool->lock));
    pthread_cond_destroy(&(pool->notify));

    FREE(pool);

    return 0;
}

static void *thrd_task_executor(void *arg)
{
    mixp_thread_pool_t *pool = (mixp_thread_pool_t *)arg;
    mixp_thread_task_t task;

    for (;;)
    {
        pthread_mutex_lock(&(pool->lock));

        while ((pool->count == 0) && (!pool->shutdown))
        {
            pthread_cond_wait(&(pool->notify), &(pool->lock));
        }

        if (pool->shutdown == IMMEDIATE ||
            (pool->shutdown == GRACEFUL && pool->count == 0))
        {
            break;
        }

        task.func = pool->queue[pool->head].func;
        task.args = pool->queue[pool->head].args;

        pool->head += 1;
        pool->head = (pool->head == pool->size) ? 0 : pool->head;
        pool->count -= 1;

        pthread_mutex_unlock(&(pool->lock));

        (*(task.func))(task.args);
    }

    pool->started--;
    pthread_mutex_unlock(&(pool->lock));
    pthread_exit(NULL);

    return (NULL);
}